package com.bicon.botu.core.base.manager;

import com.bicon.botu.core.base.common.utils.JsonParse;
import com.bicon.botu.core.base.listener.KafkaProducerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * @author 谈东魁
 * @date 2018/1/20 17:47
 */
@Slf4j
@Service
public class SyncManager {


    /**
     * 分批发送数据到kafka队列
     *
     * @param kafkaTemplate kafka 操作对象
     * @param allList       发送的数据集合
     * @param topic         主题
     */
    public void sendKafka(KafkaTemplate kafkaTemplate, List allList, String topic) {
        sendKafka(kafkaTemplate, allList, topic, 10);
    }

    /**
     * 分批发送数据到kafka队列
     *
     * @param kafkaTemplate kafka操作对象
     * @param allList       发送的数据集合
     * @param topic         主题
     * @param batchSize     分批数据大小
     */
    public void sendKafka(KafkaTemplate kafkaTemplate, List allList, String topic, int batchSize) {
        List list = new ArrayList();
        if (CollectionUtils.isNotEmpty(allList)) {
            for (int i = 0; i < allList.size(); i++) {
                list.add(allList.get(i));
                if ((i + 1) % batchSize == 0 || i == (allList.size() - 1)) {
                    String json = JsonParse.toJson(list);
                    kafkaTemplate.send(topic, json);
                    kafkaTemplate.setProducerListener(new KafkaProducerListener(kafkaTemplate));
                    list.clear();
                }
            }
        }
    }
}
